package com.academic.web.controller.scala;

import lombok.extern.slf4j.Slf4j;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.functions;
import org.apache.spark.sql.expressions.UserDefinedFunction;
import org.apache.spark.sql.types.DataTypes;

import java.util.List;
import java.util.Map;

@Slf4j
public class AcademicAnalysis {

    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder()
                .appName("AcademicAnalysis")
                .master("local") // 本地模式，实际部署时可以使用YARN或Mesos
                .getOrCreate();

        String hdfsPath = "hdfs://192.168.18.139:8020/upload/1730890373486/计算机科学与技术-181班-成绩信息(1).xls";
        Dataset<Row> data = readDataFromHDFS(spark, hdfsPath);

        // 计算成绩指标、学习行为指标和课程难度指标
        Dataset<Row> academicMetrics = calculateAcademicMetrics(data);

        // 生成预警等级
        generateWarningLevels(academicMetrics);

        // 使用模型进行预测分析
        // ...（此处可以使用随机森林或线性回归模型）

        spark.stop();
    }

    private static Dataset<Row> readDataFromHDFS(SparkSession spark, String hdfsPath) {
        return spark.read()
                .option("header", "true")
                .option("inferSchema", "true")
                .csv(hdfsPath);
    }

    private static Dataset<Row> calculateAcademicMetrics(Dataset<Row> data) {
        // 计算平时成绩、考试成绩、成绩变化趋势、绩点（GPA）等
        // ...（此处添加具体的计算逻辑）

        // 例如：计算绩点（GPA）
        UserDefinedFunction calculateGPA = functions.udf((Double totalScore, Double totalCredits) -> {
            return totalScore / totalCredits;
        }, DataTypes.DoubleType);

        data = data.withColumn("GPA", calculateGPA.apply(functions.col("总成绩"), functions.col("学分")));

        return data;
    }

    private static void generateWarningLevels(Dataset<Row> academicMetrics) {
        // 生成预警等级
        // ...（此处添加具体的生成预警等级逻辑）

        // 例如：根据GPA生成预警等级
        academicMetrics.createOrReplaceTempView("academic_metrics");
        Dataset<Row> warningLevels = academicMetrics.sqlContext().sql(
                "SELECT 学号, 姓名, GPA, " +
                        "CASE " +
                        "   WHEN GPA < 2.0 THEN '红色预警' " +
                        "   WHEN GPA < 3.0 THEN '黄色预警' " +
                        "   ELSE '蓝色预警' " +
                        "END AS 预警等级 " +
                        "FROM academic_metrics"
        );

        warningLevels.show();
    }
}
